package com.zzy.study.socketANDnio;

import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
 * 失败的作品
 * server端多线程
 * acceptSelector处理接受数据
 * selector处理读写数据
 * @author eason
 *
 */
public class NIOServerDoubleThread {
	
	private static final Log LOG = LogFactory.getLog(NIOServerDoubleThread.class);
	
	private ServerSocketChannel serverChannel;
	private SocketChannel socketChannel;
	private Selector acceptSelector;
	private static final int BLOCK_SIZE = 1024;
	private ByteBuffer receiveBuffer =  ByteBuffer.allocate(BLOCK_SIZE);
	private ByteBuffer sendBuffer = ByteBuffer.allocate(BLOCK_SIZE);
	private int count;
	private int flag = 0;
	private static final String SEND_STRING = "message from server ";
	
	private BlockingQueue<SocketChannel> QUEUE = new LinkedBlockingQueue<SocketChannel>();
    private List<ContextChangedListener> listeners = new ArrayList<ContextChangedListener>();


	public static void main(String[] args) throws IOException {
		new NIOServerDoubleThread(9111).start();
	}

	public NIOServerDoubleThread(int port) throws IOException {
		init(port);
		
		Handle handle = new Handle();
		listeners.add(handle);
		new Thread(handle).start();
	}


	private void init(int port) {
		try {
			serverChannel = ServerSocketChannel.open();
			serverChannel.socket().bind(new InetSocketAddress(port));
			serverChannel.configureBlocking(false);
			acceptSelector = Selector.open();
			serverChannel.register(acceptSelector, SelectionKey.OP_ACCEPT);
			LOG.info(new String("server has start, listener port:" + port));
		} catch (IOException e) {
			LOG.error(e);
		}
	}
	
	private void start() {
		while(true) {
			try {
				if(acceptSelector.select() > 0) {
					Set<SelectionKey> selectionKeys = acceptSelector.selectedKeys();
					Iterator iterator = selectionKeys.iterator();
					while(iterator.hasNext()) {
						SelectionKey selectionKey = (SelectionKey) iterator.next();
						iterator.remove();
						
						if(selectionKey.isValid()){
							if(selectionKey.isAcceptable()) {
								serverChannel = (ServerSocketChannel) selectionKey.channel();
								socketChannel = serverChannel.accept();
								LOG.info(socketChannel.socket().getRemoteSocketAddress() + " has a request!");
								socketChannel.configureBlocking(false);
								QUEUE.put(socketChannel);
								for(ContextChangedListener l : listeners) {
									l.doChange();
									acceptSelector.wakeup();
								}
								LOG.info("socketChannel:" + socketChannel + " has add to queue");
//								socketChannel.register(selector, SelectionKey.OP_READ);
							}
						}
							
					}
				}
			} catch (IOException e) {
				LOG.error(e);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}


	private void handleKey(SelectionKey selectionKey, Selector selector) {
		try{
			if(selectionKey.isReadable()) {
				socketChannel = (SocketChannel) selectionKey.channel();
				socketChannel.configureBlocking(false);
    			receiveBuffer.clear();
				count = socketChannel.read(receiveBuffer);
				if(count > 0) {
					LOG.info(new String(receiveBuffer.array(), 0, count) + " has received from client!");
					socketChannel.register(selector, SelectionKey.OP_WRITE);
				}
			}else if(selectionKey.isWritable()) {
				socketChannel = (SocketChannel) selectionKey.channel();  
    			sendBuffer.clear();  
    			String msg = SEND_STRING + flag++;
    			
    			ByteArrayOutputStream response = new ByteArrayOutputStream(10240);
    			DataOutputStream out = new DataOutputStream(response);
    			out.writeInt(msg.length());
    			out.write(msg.getBytes());
    			
    			sendBuffer.put(ByteBuffer.wrap(response.toByteArray()));
    			sendBuffer.flip();  
    			socketChannel.write(sendBuffer);  
    			LOG.info(new String(msg + " has set to client!"));
    			socketChannel.register(selector, SelectionKey.OP_READ);  
			}
		}catch (Exception e) {
			selectionKey.cancel();
			if (selectionKey.channel() != null){
				try {
					selectionKey.channel().close();
				} catch (IOException e1) {
					LOG.error(e);
				}
			}
		}
	}
	
	private class Handle implements Runnable, ContextChangedListener {
		private SocketChannel socketChannel;
		private Selector selector ;
		private boolean READY;
		
		public Handle() {
			try {
				selector = Selector.open();
			} catch (IOException e) {
				e.printStackTrace();
			}
			READY = false;
		}
		
		public void run() {
			while(true) {
				try {
					if(READY) {
						socketChannel = QUEUE.take();
						if(socketChannel != null) {
							LOG.info("socketChannel:" + socketChannel + " has take from queue");
							socketChannel.configureBlocking(false);
							socketChannel.register(selector, SelectionKey.OP_READ);
						}
							
						selector.select();
						Set<SelectionKey> selectionKeys = selector.selectedKeys();
						Iterator iterator = selectionKeys.iterator();
						while(iterator.hasNext()) {
							SelectionKey selectionKey = (SelectionKey) iterator.next();
							iterator.remove();
							
							if(selectionKey.isValid()){
								handleKey(selectionKey, selector);
							}
								
						}
					}
					READY = false;
					
					
					
				} catch (IOException e) {
					e.printStackTrace();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}

		public void setREADY(boolean rEADY) {
			READY = rEADY;
		}

		@Override
		public void doChange() {
			READY = true;
		}
	}
	
}
